package com.neusoft.databus.agent.pool.socket2;

import java.net.Socket;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.neusoft.databus.agent.pool.socket.SocketConnector;
import com.neusoft.databus.common.connection.AbstractConnection;
import com.neusoft.databus.common.connection.SocketConnection;
import com.neusoft.databus.common.exception.DisconnectException;
import com.neusoft.databus.common.exception.RetryableException;
import com.neusoft.databus.common.pool.Connector;

import lombok.extern.slf4j.Slf4j;

/**
 * description
 *
 * @author sunchf
 * @date 2018年12月13日 下午3:10:17
 */
@Slf4j
@Component
public class Socket2Connector implements Connector<Socket> {

    @Autowired
    Socket2Pool pool;

    final Map<String, ReadWriteLock> lockMap = new ConcurrentHashMap<>();

    @Override
    public Socket getConn(final AbstractConnection connection) {
        return this.getConn(connection, 0);
    }

    @Override
    public Socket getConn(final AbstractConnection connection, final int timeout) {
        final SocketConnection socketConnection = (SocketConnection) connection;
        final String lockId =
                socketConnection.getIpAddress() + ":" + socketConnection.getPort() + ":"
                        + socketConnection.getSeq();
        Socket socket = null;
        try {
            synchronized (SocketConnector.class) {
                ReadWriteLock lock = this.lockMap.get(lockId);
                if (null == lock) {
                    lock = new ReentrantReadWriteLock(true);
                    this.lockMap.put(lockId, lock);
                }
            }

            final ReadWriteLock lock = this.lockMap.get(lockId);
            lock.writeLock().lock();
            socket = this.pool.borrowObject(socketConnection);
        } catch (final NoSuchElementException e) {
            throw new RetryableException(e.getMessage(), e);
        } catch (final Exception e) {
            // 通过对象池Factory创建新对象时，抛出的所有异常，都会在这里捕捉到。
            // 封装成中断异常
            throw new DisconnectException(e.getMessage(), e);
        } finally {
            final ReadWriteLock lock = this.lockMap.get(lockId);
            lock.writeLock().unlock();
        }
        return socket;
    }

    @Override
    public void returnConn(final AbstractConnection connection, final Socket socket) {
        try {
            this.pool.returnObject((SocketConnection) connection, socket);
        } catch (final Exception e) {
            Socket2Connector.log.warn("归还Socket2对象时错误, {}", e.getMessage());
        }
    }

    @Override
    @Deprecated
    public void invalidateConn(final AbstractConnection connection, final Socket socket) {
        if (null == socket) {
            return;
        }
        try {
            this.pool.invalidateObject((SocketConnection) connection, socket);
            Socket2Connector.log.info("invalidate pooled socket {}", connection.toString());
        } catch (final Exception e) {
            Socket2Connector.log.error(e.getMessage(), e);
        }
    }

    @Override
    @Deprecated
    public void removeAllConns(final AbstractConnection connection) {
        try {
            this.pool.clear((SocketConnection) connection);
            Socket2Connector.log.info("clear pooled socket {}", connection.toString());
        } catch (final Exception e) {
            Socket2Connector.log.error(e.getMessage(), e);
        }

    }

}
